package com.orangeandbronze.demos.mapreduce;

import java.nio.ByteBuffer;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;

import com.google.appengine.tools.mapreduce.Reducer;
import com.google.appengine.tools.mapreduce.ReducerInput;

public class IndexReducer extends Reducer<String, String, ByteBuffer> {

	private static final long serialVersionUID = 1664448285248783646L;

	/*
	 * Based on org.apache.commons.lang.StringUtils.
	 */
	private String join(Iterable<String> iterable, String separator) {
		final Iterator<String> iterator = iterable.iterator();
		StringBuilder buffer = new StringBuilder();
		if (iterator.hasNext()) {
			buffer.append(iterator.next()); // first
		}
		while (iterator.hasNext()) {
			buffer.append(separator);
			buffer.append(iterator.next());
		}
		return buffer.toString();
	}

	@Override
	public void reduce(String key, ReducerInput<String> values) {
		Set<String> filenameSet = new HashSet<String>();
		while (values.hasNext()) {
			filenameSet.add(values.next());
		}
		getContext().emit(ByteBuffer.wrap(
				(key + ": [" + join(filenameSet, ", ") + "]\n").getBytes()));
	}

	/*
def index_map(data):
  """Index demo map function."""
  (entry, text_fn) = data
  text = text_fn()

  logging.debug("Got %s", entry.filename)
  for s in split_into_sentences(text):
    for w in split_into_words(s.lower()):
      yield (w, entry.filename)


def index_reduce(key, values):
  """Index demo reduce function."""
  yield "%s: %s\n" % (key, list(set(values)))
  	 */
}
